package com.boarsoft.boar.batch.quartz;

import java.sql.Timestamp;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import com.boarsoft.boar.batch.service.BatchInfoService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.boarsoft.boar.batch.biz.BatchInstBiz;
import com.boarsoft.boar.batch.biz.BatchLockBiz;
import com.boarsoft.boar.batch.entity.BatchInst;
import com.boarsoft.boar.batch.entity.BatchLock;
import com.boarsoft.boar.batch.service.BatchInstService;
import com.boarsoft.boar.batch.service.BatchScheduler;
import com.boarsoft.common.Util;
import com.boarsoft.common.util.InetUtil;

public class QuartzSchedulerMonitor implements Runnable, BatchInstService {
	private final static Logger log = LoggerFactory.getLogger(QuartzSchedulerMonitor.class);

	protected BatchScheduler batchScheduler;

	@Autowired
	protected BatchInstBiz instBiz;

	@Autowired
	protected BatchLockBiz lockBiz;

	protected ScheduledExecutorService scheduler;

	protected BatchScheduler broadcast;

	protected BatchInst inst;

	protected String addr;

	protected String catalog = "default";

	protected long delay = 5L;

	protected long initDelay = 0L;

	protected int timeout = 20;// in seconds

	@Autowired
	private BatchInfoService infoSvc;
	private static final String JOBID_TIMEOUTCHECK = "2";
	private static final String JOBID_RUNSTATUSCHECK = "1";

	@PostConstruct
	public void init() {
		// 尝试注册自己到BatchInst表
		if (inst != null) {
			addr = inst.getAddr();
			catalog = inst.getCatalog();
		}
		if (Util.strIsEmpty(addr)) {
			addr = InetUtil.getAddr();
		}
		if (Util.strIsEmpty(catalog)) {
			catalog = addr;// 自己单独为一组
		}
		inst = instBiz.get(addr);
		if (inst == null || !catalog.equals(inst.getCatalog())) {
			// 若 库中不存在对应的数据，或组别不一致，则 插入、更新新数据
			BatchInst nu = new BatchInst();
			nu.setAddr(addr);
			nu.setCatalog(catalog);
			// 初始值为2018-01-01 00:00:00.0
			nu.setLockTime(new Timestamp(1514736000000L));
			nu.setLastTime(new Timestamp(1514736000000L));
			instBiz.save(nu);// merge
		}
		// 启动定时任务:批量实例锁竞争
		scheduler.scheduleWithFixedDelay(this, initDelay, delay, TimeUnit.SECONDS);
	}

	@Override
	public void run() {
		//自动注册定时任务
		infoSvc.enable(JOBID_RUNSTATUSCHECK); //执行状态更新定时任务
		infoSvc.enable(JOBID_TIMEOUTCHECK);   //超时检查定时任务

		// 获取锁，如果成功启动定时器，否则停止定时器
		if (this.checkLock()) {
			// 如果获取锁成功，且定时器为未启动，或处理于standby状态，则启动它
			if (!batchScheduler.isStarted() || batchScheduler.isStandByMode()) {
				try {
					log.warn("Switch current quartz instance as master node, start scheduler");
					batchScheduler.start();
				} catch (Exception e) {
					log.error("Error on start quartz scheduler", e);
				}
			}
		} else {
			// log.info("锁定BatchLock失败");
			if (!batchScheduler.isStandByMode()) {
				batchScheduler.stop();
				log.warn("Switch current quartz instance as backup node");
			}
		}
	}

	protected boolean checkLock() {
		// 以批量集群group做主键获取BatchLock，作为乐观锁，用于锁定对BatchInst表属于该集群的全部记录的lockTime字段的更新操作
		BatchLock ol = lockBiz.get(catalog);
		if (ol == null) {
			try {
				BatchLock nl = new BatchLock();
				nl.setCode(catalog);
				nl.setAddr(addr);
				nl.setVersion(0L);
//				nl.setStatus(BatchLock.STATUS_DISABLE);
				nl.setStatus(BatchLock.STATUS_ENABLE);
				lockBiz.save(nl);
				ol = nl;
			} catch (Exception e) {
				log.error("Init BatchLock failed", e);
				return false; // 多个实例可能多次插入记录，造成主键重复错误。捕获错误，并直接返回失败。
			}
		}
		// 批量集群处于运行状态
		if (BatchLock.STATUS_ENABLE == ol.getStatus()) {
			// 获取“运行实例”（BatchInst中lockTime最新且未超时的数据）
			BatchInst activeLock = instBiz.getActiveLock(timeout);
			// log.info("Current active broker is {}", activeLock.getAddr());
			// 检查：“运行实例”是否不存在或正是自己，则尝试获取或更新锁
			if (activeLock == null || addr.equals(activeLock.getAddr())) {
				// 若注册自己为“运行实例”成功，一并更新锁时间和签到时间LastTime
				if (instBiz.acquireLock(addr, timeout, ol)) {
					return true;
				}
			}
		} else {
			log.warn("Batch broker cluster {} is disabled", catalog);
		}
		// 若，注册自己为“运行实例”不成功，更新签到时间LastTime
		instBiz.updateLastTime(addr);
		return false;
	}

	@PreDestroy
	public void destroy() {
		try {
			this.scheduler.shutdown();// 关闭定时线程
		} catch (Exception e) {
			log.error("Error on shutdown scheduler thread pool", e);
		}
	}

	public void enableGroup() {
		lockBiz.updateStatus(catalog, BatchLock.STATUS_ENABLE);
	}

	public void disableGroup() {
		lockBiz.updateStatus(catalog, BatchLock.STATUS_DISABLE);
	}

	public boolean groupStatus() {
		BatchLock lock = lockBiz.get(catalog);
		if (BatchLock.STATUS_ENABLE == lock.getStatus())
			return true;
		return false;
	}

	@Override
	public Object start(BatchInst batchInst) {
		log.info("启动目标调度[{}]", batchInst.getAddr());
		if (this.inst.getAddr().equals(batchInst.getAddr())) {
			// 若当前调度的地址和组别与广播消息中的一致，则启动，尝试成为主调度
			// batchScheduler.start();
			// log.info("设置为主调度成功");
			run(); // 尝试启动。不直接启动调度，避免与自动检查机制冲突。
		}
		return true;
	}

	@Override
	public Object stop(BatchInst batchInst) {
		log.info("停止目标调度[{}]", batchInst.getAddr());
		if (this.inst.getAddr().equals(batchInst.getAddr())) {
			batchScheduler.stop();
		}
		return true;
	}

	@Override
	public Object stopAll() {
		log.info("停止本机全部调度");
		batchScheduler.stop();
		return true;
	}

	public BatchScheduler getBatchScheduler() {
		return batchScheduler;
	}

	public void setBatchScheduler(BatchScheduler batchScheduler) {
		this.batchScheduler = batchScheduler;
	}

	public BatchLockBiz getLockBiz() {
		return lockBiz;
	}

	public void setLockBiz(BatchLockBiz lockBiz) {
		this.lockBiz = lockBiz;
	}

	public ScheduledExecutorService getScheduler() {
		return scheduler;
	}

	public void setScheduler(ScheduledExecutorService scheduler) {
		this.scheduler = scheduler;
	}

	public long getDelay() {
		return delay;
	}

	public void setDelay(long delay) {
		this.delay = delay;
	}

	public long getInitDelay() {
		return initDelay;
	}

	public void setInitDelay(long initDelay) {
		this.initDelay = initDelay;
	}

	public int getTimeout() {
		return timeout;
	}

	public void setTimeout(int timeout) {
		this.timeout = timeout;
	}

	public BatchInst getInst() {
		return inst;
	}

	public void setInst(BatchInst inst) {
		this.inst = inst;
	}

	public String getAddr() {
		return addr;
	}

	public void setAddr(String addr) {
		this.addr = addr;
	}

	public String getCatalog() {
		return catalog;
	}

	public void setCatalog(String catalog) {
		this.catalog = catalog;
	}

}